<!DOCTYPE html>
<html>
<head><meta name="generator" content="Hexo 3.8.0">
  <meta charset="utf-8">
  
  <title>rocketmq | JAVA</title>
  <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
  
  <meta name="keywords" content="MQ">
  
  
  
  
  <meta name="description" content="MQ有哪些作用?异步化提高系统性能系统解耦高并发削峰MQ技术选型 RocketMQ核心原理NameServer  是整个消息队列中的状态服务器，集群的各个组件通过它来了解全局的信息。每个broker都会注册到所有的NameServer上，Broker和NameServer采用心跳机制,Broker会每隔30s给所有NameServer发送心跳,告诉每个nameServer自己还活着，然后NameS">
<meta name="keywords" content="MQ">
<meta property="og:type" content="article">
<meta property="og:title" content="RocketMQ">
<meta property="og:url" content="https://jameslin23.gitee.io/2020/12/01/RocketMQ/index.html">
<meta property="og:site_name" content="JAVA">
<meta property="og:description" content="MQ有哪些作用?异步化提高系统性能系统解耦高并发削峰MQ技术选型 RocketMQ核心原理NameServer  是整个消息队列中的状态服务器，集群的各个组件通过它来了解全局的信息。每个broker都会注册到所有的NameServer上，Broker和NameServer采用心跳机制,Broker会每隔30s给所有NameServer发送心跳,告诉每个nameServer自己还活着，然后NameS">
<meta property="og:locale" content="zh-CN">
<meta property="og:image" content="https://jameslin23.gitee.io/2020/12/01/RocketMQ/image-20201203142559984.png">
<meta property="og:image" content="https://jameslin23.gitee.io/2020/12/01/RocketMQ/image-20201202202251399.png">
<meta property="og:image" content="https://jameslin23.gitee.io/2020/12/01/RocketMQ/15762560-d88bf85df93dce41.png">
<meta property="og:updated_time" content="2020-12-04T11:56:08.848Z">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="RocketMQ">
<meta name="twitter:description" content="MQ有哪些作用?异步化提高系统性能系统解耦高并发削峰MQ技术选型 RocketMQ核心原理NameServer  是整个消息队列中的状态服务器，集群的各个组件通过它来了解全局的信息。每个broker都会注册到所有的NameServer上，Broker和NameServer采用心跳机制,Broker会每隔30s给所有NameServer发送心跳,告诉每个nameServer自己还活着，然后NameS">
<meta name="twitter:image" content="https://jameslin23.gitee.io/2020/12/01/RocketMQ/image-20201203142559984.png">
  
    <link rel="alternate" href="/atom.xml" title="JAVA" type="application/atom+xml">
  

  

  <link rel="icon" href="/css/images/mylogo.jpg">
  <link rel="apple-touch-icon" href="/css/images/mylogo.jpg">
  
    <link href="//fonts.googleapis.com/css?family=Source+Code+Pro" rel="stylesheet" type="text/css">
  
  <link href="https://fonts.googleapis.com/css?family=Open+Sans|Montserrat:700" rel="stylesheet" type="text/css">
  <link href="https://fonts.googleapis.com/css?family=Roboto:400,300,300italic,400italic" rel="stylesheet" type="text/css">
  <link href="//netdna.bootstrapcdn.com/font-awesome/4.0.3/css/font-awesome.css" rel="stylesheet">
  <style type="text/css">
    @font-face{font-family:futura-pt; src:url("/css/fonts/FuturaPTBold.otf") format("woff");font-weight:500;font-style:normal;}
    @font-face{font-family:futura-pt-light; src:url("/css/fonts/FuturaPTBook.otf") format("woff");font-weight:lighter;font-style:normal;}
    @font-face{font-family:futura-pt-italic; src:url("/css/fonts/FuturaPTBookOblique.otf") format("woff");font-weight:400;font-style:italic;}
}

  </style>
  <link rel="stylesheet" href="/css/style.css">

  <script src="/js/jquery-3.1.1.min.js"></script>
  <script src="/js/bootstrap.js"></script>

  <!-- Bootstrap core CSS -->
  <link rel="stylesheet" href="/css/bootstrap.css">

  
    <link rel="stylesheet" href="/css/dialog.css">
  

  

  
    <link rel="stylesheet" href="/css/header-post.css">
  

  
  
  
    <link rel="stylesheet" href="/css/vdonate.css">
  

</head>
</html>


  <body data-spy="scroll" data-target="#toc" data-offset="50">


  
  <div id="container">
    <div id="wrap">
      
        <header>

    <div id="allheader" class="navbar navbar-default navbar-static-top" role="navigation">
        <div class="navbar-inner">
          
          <div class="container"> 
            <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".navbar-collapse">
              <span class="sr-only">Toggle navigation</span>
              <span class="icon-bar"></span>
              <span class="icon-bar"></span>
              <span class="icon-bar"></span>
            </button>

            
              <a class="brand" style="border-width: 0;">
                <p>JAVA</p>
              </a>
            
            
            <div class="navbar-collapse collapse">
              <ul class="hnav navbar-nav">
                
                  <li> <a class="main-nav-link" href="/">首页</a> </li>
                
                  <li> <a class="main-nav-link" href="/archives">归档</a> </li>
                
                  <li> <a class="main-nav-link" href="/categories">分类</a> </li>
                
                  <li> <a class="main-nav-link" href="/tags">标签</a> </li>
                
                  <li> <a class="main-nav-link" href="/about">关于</a> </li>
                
                  <li><div id="search-form-wrap">

    <form class="search-form">
        <input type="text" class="ins-search-input search-form-input" placeholder>
        <button type="submit" class="search-form-submit"></button>
    </form>
    <div class="ins-search">
    <div class="ins-search-mask"></div>
    <div class="ins-search-container">
        <div class="ins-input-wrapper">
            <input type="text" class="ins-search-input" placeholder="请输入关键词...">
            <span class="ins-close ins-selectable"><i class="fa fa-times-circle"></i></span>
        </div>
        <div class="ins-section-wrapper">
            <div class="ins-section-container"></div>
        </div>
    </div>
</div>
<script>
(function (window) {
    var INSIGHT_CONFIG = {
        TRANSLATION: {
            POSTS: '文章',
            PAGES: '页面',
            CATEGORIES: '分类',
            TAGS: '标签',
            UNTITLED: '(无标题)',
        },
        ROOT_URL: '/',
        CONTENT_URL: '/content.json',
    };
    window.INSIGHT_CONFIG = INSIGHT_CONFIG;
})(window);
</script>
<script src="/js/insight.js"></script>

</div></li>
            </ul></div>
          </div>
                
      </div>
    </div>

</header>



      
            
      <div id="content" class="outer">
        
          <section id="main" style="float:none;"><article id="post-RocketMQ" style="width: 75%; float:left;" class="article article-type-post" itemscope itemprop="blogPost">
  <div id="articleInner" class="article-inner">
    
    
      <header class="article-header">
        
  
    <h1 class="thumb" itemprop="name">
      RocketMQ
    </h1>
  

      </header>
    
    <div class="article-meta">
      
	<a href="/2020/12/01/RocketMQ/" class="article-date">
	  <time datetime="2020-12-01T00:30:33.000Z" itemprop="datePublished">2020-12-01</time>
	</a>

      
    <a class="article-category-link" href="/categories/消息队列/">消息队列</a>

      
	<a class="article-views">
	<span id="busuanzi_container_page_pv">
		阅读量<span id="busuanzi_value_page_pv"></span>
	</span>
	</a>

      

    </div>
    <div class="article-entry" itemprop="articleBody">
      
        <h1 id="MQ有哪些作用"><a href="#MQ有哪些作用" class="headerlink" title="MQ有哪些作用?"></a>MQ有哪些作用?</h1><h2 id="异步化提高系统性能"><a href="#异步化提高系统性能" class="headerlink" title="异步化提高系统性能"></a>异步化提高系统性能</h2><h2 id="系统解耦"><a href="#系统解耦" class="headerlink" title="系统解耦"></a>系统解耦</h2><h2 id="高并发削峰"><a href="#高并发削峰" class="headerlink" title="高并发削峰"></a>高并发削峰</h2><h1 id="MQ技术选型"><a href="#MQ技术选型" class="headerlink" title="MQ技术选型"></a>MQ技术选型</h1><p><img src="https://jameslin23.gitee.io/2020/12/01/RocketMQ/image-20201203142559984.png" alt="image-20201203142559984"></p>
<h1 id="RocketMQ核心原理"><a href="#RocketMQ核心原理" class="headerlink" title="RocketMQ核心原理"></a>RocketMQ核心原理</h1><h2 id="NameServer"><a href="#NameServer" class="headerlink" title="NameServer"></a>NameServer</h2><blockquote>
<p> 是整个消息队列中的状态服务器，集群的各个组件通过它来了解全局的信息。每个broker都会注册到所有的NameServer上，Broker和NameServer采用心跳机制,Broker会每隔30s给所有NameServer发送心跳,告诉每个nameServer自己还活着，然后NameServer会每隔10s运行一个任务，去检测一下各个Broker的最近一次心跳时间，如果某个Broker超过120s都没发心跳了，就认为这个Broker已经挂掉了。</p>
<p>发送消息。通过nameServer去获取路由信息，就知道集群有哪些Broker等消息<br>消费消息，也会到NameServer获取路由信息，去找到对应的Broker获取消息</p>
</blockquote>
<h2 id="Broker"><a href="#Broker" class="headerlink" title="Broker"></a>Broker</h2><p>真正存储数据的地方。</p>
<h3 id="CommitLog"><a href="#CommitLog" class="headerlink" title="CommitLog"></a>CommitLog</h3><blockquote>
<p>消息内容原文的存储文件，同Kafka一样，消息是变长的，顺序写入</p>
<p>生成规则:</p>
<p>每个文件的默认1G =1024 * 1024 * 1024，commitlog的文件名fileName，名字长度为20位，左边补零，剩余为起始偏移量；比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1 073 741 824Byte；当这个文件满了，第二个文件名字为00000000001073741824，起始偏移量为1073741824, 消息存储的时候会顺序写入文件，当文件满了则写入下一个文件</p>
</blockquote>
<h3 id="ConsumeQueue"><a href="#ConsumeQueue" class="headerlink" title="ConsumeQueue"></a>ConsumeQueue</h3><blockquote>
<p>ConsumeQueue中并不需要存储消息的内容，而存储的是消息在CommitLog中的offset。也就是说，ConsumeQueue其实是CommitLog的一个索引文件。</p>
<p>$HOME/store/consumequeue/{topic}/{queueId}/{fileName}</p>
</blockquote>
<p>ConsumeQueue是定长的结构，每1条记录固定的20个字节。很显然，Consumer消费消息的时候，要读2次：先读ConsumeQueue得到offset，再读CommitLog得到消息内容</p>
<p><img src="https://jameslin23.gitee.io/2020/12/01/RocketMQ/image-20201202202251399.png" alt="image-20201202202251399"></p>
<p>  <strong>ConsumeQueue的作用</strong></p>
<ol>
<li>通过broker保存的offset可以在ConsumeQueue中获取消息，从而快速的定位到commitLog的消息位置</li>
<li>过滤tag是也是通过遍历ConsumeQueue来实现的（先比较hash(tag)符合条件的再到consumer比较tag原文）</li>
<li>并且ConsumeQueue还能保存于操作系统的PageCache进行缓存提升检索性能</li>
</ol>
<blockquote>
<p> 当你的Broker收到一条消息写入了CommitLog之后，其实他同时会将这条消息在CommitLog中的物理位置，也就是一个文<br>件偏移量，就是一个offset，写入到这条消息所属的MessageQueue对应的ConsumeQueue文件中去。</p>
</blockquote>
<h3 id="异步刷盘和同步刷盘"><a href="#异步刷盘和同步刷盘" class="headerlink" title="异步刷盘和同步刷盘"></a>异步刷盘和同步刷盘</h3><p><strong>Broker是基于OS操作系统的PageCache和顺序写两个机制，来提升写入CommitLog文件的性能的</strong></p>
<p><img src="https://jameslin23.gitee.io/2020/12/01/RocketMQ/15762560-d88bf85df93dce41.png" alt="img"></p>
<h4 id="异步刷盘"><a href="#异步刷盘" class="headerlink" title="异步刷盘"></a>异步刷盘</h4><blockquote>
<p>首先Broker是以顺序的方式将消息写入CommitLog磁盘文件的，也就是每次写入就是在文件末尾追加一条数据就可以了，对文件进行顺序写的性能要比对文件随机写的性能提升很多</p>
<p>数据写入CommitLog文件的时候，其实不是直接写入底层的物理磁盘文件的，而是先进入OS的PageCache内存缓存中，然后后<br>续由OS的后台线程选一个时间，异步化的将OS PageCache内存缓冲中的数据刷入底层的磁盘文件</p>
<p>所以在这样的优化之下，采用<strong>磁盘文件顺序写+OS PageCache写入+OS异步刷盘</strong>的策略，基本上可以让消息写入CommitLog的性能<br>跟你直接写入内存里是差不多的，所以正是如此，才可以让Broker高吞吐的处理每秒大量的消息写入。</p>
</blockquote>
<p><strong>存在问题</strong></p>
<blockquote>
<p>在上述的异步刷盘模式下，生产者把消息发送给Broker，Broker将消息写入OS PageCache中，就直接返回ACK给生产者了。</p>
<p>生产者认为消息写入成功了，但是实际上那条消息此时是在Broker机器上的os cache中的，如果此时Broker直<br>接宕机，可能会有数据丢失的风险。</p>
</blockquote>
<h4 id="同步刷盘"><a href="#同步刷盘" class="headerlink" title="同步刷盘"></a>同步刷盘</h4><blockquote>
<p> 生产者发送一条消息出去，broker收到了消息，必须直接强制把这个消息刷入底层的物理磁盘文件中，然后才会返回ack给producer，此时你才知道消息写入成功了</p>
</blockquote>
<p><strong>优缺点</strong></p>
<blockquote>
<p>吞吐量低，但是消息不容易丢失。</p>
</blockquote>
<h2 id="Producer"><a href="#Producer" class="headerlink" title="Producer"></a>Producer</h2><p>rocketmq支持三种发送消息的方式，分别是同步发送（sync），异步发送（async）和直接发送（oneway）</p>
<h3 id="同步发送（Sync）"><a href="#同步发送（Sync）" class="headerlink" title="同步发送（Sync）"></a>同步发送（Sync）</h3><blockquote>
<p>这种只有在消息完全发送完成之后才返回结果，此方法需要同步等待发送结果的时间为代价</p>
<p>这种方式具有内部重试机制,即在主动声明本次发送失败之前，内部实现将重试一定的次数，默认为2次（DefaultMQProducer＃setRetryTimesWhenSendFailed（））发送结果存在同一个消息可能被多次发送给broker,这里需要应用开发者在自己消费端处理。</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br></pre></td><td class="code"><pre><span class="line"></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SyncProducer</span> </span>&#123;</span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">    	<span class="comment">// 实例化消息生产者Producer</span></span><br><span class="line">        DefaultMQProducer producer = <span class="keyword">new</span> DefaultMQProducer(<span class="string">"please_rename_unique_group_name"</span>);</span><br><span class="line">    	<span class="comment">// 设置NameServer的地址</span></span><br><span class="line">    	producer.setNamesrvAddr(<span class="string">"localhost:9876"</span>);</span><br><span class="line">    	<span class="comment">// 启动Producer实例</span></span><br><span class="line">        producer.start();</span><br><span class="line">    	<span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">100</span>; i++) &#123;</span><br><span class="line">    	    <span class="comment">// 创建消息，并指定Topic，Tag和消息体</span></span><br><span class="line">    	    Message msg = <span class="keyword">new</span> Message(<span class="string">"TopicTest"</span> <span class="comment">/* Topic */</span>,</span><br><span class="line">        	<span class="string">"TagA"</span> <span class="comment">/* Tag */</span>,</span><br><span class="line">        	(<span class="string">"Hello RocketMQ "</span> + i).getBytes(RemotingHelper.DEFAULT_CHARSET) <span class="comment">/* Message body */</span></span><br><span class="line">        	);</span><br><span class="line">        	<span class="comment">// 发送消息到一个Broker</span></span><br><span class="line">            SendResult sendResult = producer.send(msg);</span><br><span class="line">            <span class="comment">// 通过sendResult返回消息是否成功送达</span></span><br><span class="line">            System.out.printf(<span class="string">"%s%n"</span>, sendResult);</span><br><span class="line">    	&#125;</span><br><span class="line">    	<span class="comment">// 如果不再发送消息，关闭Producer实例。</span></span><br><span class="line">    	producer.shutdown();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="异步发送（async）"><a href="#异步发送（async）" class="headerlink" title="异步发送（async）"></a>异步发送（async）</h3><blockquote>
<p>异步消息通常在对响应时间敏感的业务场景，即发送端不能容忍长时间等待Broker的响应,异步模式也在内部实现了重试机制，默认次数为2次(setRetryTimesWhenSendAsyncFailed()</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">AsyncProducer</span> </span>&#123;</span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">    	<span class="comment">// 实例化消息生产者Producer</span></span><br><span class="line">        DefaultMQProducer producer = <span class="keyword">new</span> DefaultMQProducer(<span class="string">"please_rename_unique_group_name"</span>);</span><br><span class="line">    	<span class="comment">// 设置NameServer的地址</span></span><br><span class="line">        producer.setNamesrvAddr(<span class="string">"localhost:9876"</span>);</span><br><span class="line">    	<span class="comment">// 启动Producer实例</span></span><br><span class="line">        producer.start();</span><br><span class="line">        producer.setRetryTimesWhenSendAsyncFailed(<span class="number">0</span>);</span><br><span class="line">    	<span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">100</span>; i++) &#123;</span><br><span class="line">                <span class="keyword">final</span> <span class="keyword">int</span> index = i;</span><br><span class="line">            	<span class="comment">// 创建消息，并指定Topic，Tag和消息体</span></span><br><span class="line">                Message msg = <span class="keyword">new</span> Message(<span class="string">"TopicTest"</span>,</span><br><span class="line">                    <span class="string">"TagA"</span>,</span><br><span class="line">                    <span class="string">"OrderID188"</span>,</span><br><span class="line">                    <span class="string">"Hello world"</span>.getBytes(RemotingHelper.DEFAULT_CHARSET));</span><br><span class="line">                <span class="comment">// SendCallback接收异步返回结果的回调</span></span><br><span class="line">                producer.send(msg, <span class="keyword">new</span> SendCallback() &#123;</span><br><span class="line">                    <span class="meta">@Override</span></span><br><span class="line">                    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onSuccess</span><span class="params">(SendResult sendResult)</span> </span>&#123;</span><br><span class="line">                        System.out.printf(<span class="string">"%-10d OK %s %n"</span>, index,</span><br><span class="line">                            sendResult.getMsgId());</span><br><span class="line">                    &#125;</span><br><span class="line">                    <span class="meta">@Override</span></span><br><span class="line">                    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onException</span><span class="params">(Throwable e)</span> </span>&#123;</span><br><span class="line">      	              System.out.printf(<span class="string">"%-10d Exception %s %n"</span>, index, e);</span><br><span class="line">      	              e.printStackTrace();</span><br><span class="line">                    &#125;</span><br><span class="line">            	&#125;);</span><br><span class="line">    	&#125;</span><br><span class="line">    	<span class="comment">// 如果不再发送消息，关闭Producer实例。</span></span><br><span class="line">    	producer.shutdown();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="直接发送-sendOneway"><a href="#直接发送-sendOneway" class="headerlink" title="直接发送(sendOneway)"></a>直接发送(sendOneway)</h3><blockquote>
<p> 这种方式主要用在不特别关心发送结果的场景，例如日志发送。</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">OnewayProducer</span> </span>&#123;</span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception</span>&#123;</span><br><span class="line">    	<span class="comment">// 实例化消息生产者Producer</span></span><br><span class="line">        DefaultMQProducer producer = <span class="keyword">new</span> DefaultMQProducer(<span class="string">"please_rename_unique_group_name"</span>);</span><br><span class="line">    	<span class="comment">// 设置NameServer的地址</span></span><br><span class="line">        producer.setNamesrvAddr(<span class="string">"localhost:9876"</span>);</span><br><span class="line">    	<span class="comment">// 启动Producer实例</span></span><br><span class="line">        producer.start();</span><br><span class="line">    	<span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">100</span>; i++) &#123;</span><br><span class="line">        	<span class="comment">// 创建消息，并指定Topic，Tag和消息体</span></span><br><span class="line">        	Message msg = <span class="keyword">new</span> Message(<span class="string">"TopicTest"</span> <span class="comment">/* Topic */</span>,</span><br><span class="line">                <span class="string">"TagA"</span> <span class="comment">/* Tag */</span>,</span><br><span class="line">                (<span class="string">"Hello RocketMQ "</span> + i).getBytes(RemotingHelper.DEFAULT_CHARSET) <span class="comment">/* Message body */</span></span><br><span class="line">        	);</span><br><span class="line">        	<span class="comment">// 发送单向消息，没有任何返回结果</span></span><br><span class="line">        	producer.sendOneway(msg);</span><br><span class="line"> </span><br><span class="line">    	&#125;</span><br><span class="line">    	<span class="comment">// 如果不再发送消息，关闭Producer实例。</span></span><br><span class="line">    	producer.shutdown();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="顺序消息发送"><a href="#顺序消息发送" class="headerlink" title="顺序消息发送"></a>顺序消息发送</h3><blockquote>
<p>消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。 顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。</p>
<p>要保证发送与消费有序，那同一类型的消息-发送到同一个队列当中，同一个队列只能一个线程去处理，即同一个消费者。</p>
<p>面用订单进行分区有序的示例。一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> org.apache.rocketmq.client.producer.DefaultMQProducer;</span><br><span class="line"><span class="keyword">import</span> org.apache.rocketmq.client.producer.MessageQueueSelector;</span><br><span class="line"><span class="keyword">import</span> org.apache.rocketmq.client.producer.SendResult;</span><br><span class="line"><span class="keyword">import</span> org.apache.rocketmq.common.message.Message;</span><br><span class="line"><span class="keyword">import</span> org.apache.rocketmq.common.message.MessageQueue;</span><br><span class="line"> </span><br><span class="line"><span class="keyword">import</span> java.text.SimpleDateFormat;</span><br><span class="line"><span class="keyword">import</span> java.util.ArrayList;</span><br><span class="line"><span class="keyword">import</span> java.util.Date;</span><br><span class="line"><span class="keyword">import</span> java.util.List;</span><br><span class="line"> </span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">* Producer，发送顺序消息</span></span><br><span class="line"><span class="comment">*/</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Producer</span> </span>&#123;</span><br><span class="line"> </span><br><span class="line">   <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">       DefaultMQProducer producer = <span class="keyword">new</span> DefaultMQProducer(<span class="string">"please_rename_unique_group_name"</span>);</span><br><span class="line"> </span><br><span class="line">       producer.setNamesrvAddr(<span class="string">"127.0.0.1:9876"</span>);</span><br><span class="line"> </span><br><span class="line">       producer.start();</span><br><span class="line"> </span><br><span class="line">       String[] tags = <span class="keyword">new</span> String[]&#123;<span class="string">"TagA"</span>, <span class="string">"TagC"</span>, <span class="string">"TagD"</span>&#125;;</span><br><span class="line"> </span><br><span class="line">       <span class="comment">// 订单列表</span></span><br><span class="line">       List&lt;OrderStep&gt; orderList = <span class="keyword">new</span> Producer().buildOrders();</span><br><span class="line"> </span><br><span class="line">       Date date = <span class="keyword">new</span> Date();</span><br><span class="line">       SimpleDateFormat sdf = <span class="keyword">new</span> SimpleDateFormat(<span class="string">"yyyy-MM-dd HH:mm:ss"</span>);</span><br><span class="line">       String dateStr = sdf.format(date);</span><br><span class="line">       <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">10</span>; i++) &#123;</span><br><span class="line">           <span class="comment">// 加个时间前缀</span></span><br><span class="line">           String body = dateStr + <span class="string">" Hello RocketMQ "</span> + orderList.get(i);</span><br><span class="line">           Message msg = <span class="keyword">new</span> Message(<span class="string">"TopicTest"</span>, tags[i % tags.length], <span class="string">"KEY"</span> + i, body.getBytes());</span><br><span class="line"> </span><br><span class="line">           SendResult sendResult = producer.send(msg, <span class="keyword">new</span> MessageQueueSelector() &#123;</span><br><span class="line">               <span class="meta">@Override</span></span><br><span class="line">               <span class="function"><span class="keyword">public</span> MessageQueue <span class="title">select</span><span class="params">(List&lt;MessageQueue&gt; mqs, Message msg, Object arg)</span> </span>&#123;</span><br><span class="line">                   Long id = (Long) arg;  <span class="comment">//根据订单id选择发送queue</span></span><br><span class="line">                   <span class="keyword">long</span> index = id % mqs.size();</span><br><span class="line">                   <span class="keyword">return</span> mqs.get((<span class="keyword">int</span>) index);</span><br><span class="line">               &#125;</span><br><span class="line">           &#125;, orderList.get(i).getOrderId());<span class="comment">//订单id</span></span><br><span class="line"> </span><br><span class="line">           System.out.println(String.format(<span class="string">"SendResult status:%s, queueId:%d, body:%s"</span>,</span><br><span class="line">               sendResult.getSendStatus(),</span><br><span class="line">               sendResult.getMessageQueue().getQueueId(),</span><br><span class="line">               body));</span><br><span class="line">       &#125;</span><br><span class="line"> </span><br><span class="line">       producer.shutdown();</span><br><span class="line">   &#125;</span><br></pre></td></tr></table></figure>

<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br></pre></td><td class="code"><pre><span class="line"> <span class="comment">/**</span></span><br><span class="line"><span class="comment">    * 订单的步骤</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">   <span class="keyword">private</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">OrderStep</span> </span>&#123;</span><br><span class="line">       <span class="keyword">private</span> <span class="keyword">long</span> orderId;</span><br><span class="line">       <span class="keyword">private</span> String desc;</span><br><span class="line"> </span><br><span class="line">       <span class="function"><span class="keyword">public</span> <span class="keyword">long</span> <span class="title">getOrderId</span><span class="params">()</span> </span>&#123;</span><br><span class="line">           <span class="keyword">return</span> orderId;</span><br><span class="line">       &#125;</span><br><span class="line"> </span><br><span class="line">       <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setOrderId</span><span class="params">(<span class="keyword">long</span> orderId)</span> </span>&#123;</span><br><span class="line">           <span class="keyword">this</span>.orderId = orderId;</span><br><span class="line">       &#125;</span><br><span class="line"> </span><br><span class="line">       <span class="function"><span class="keyword">public</span> String <span class="title">getDesc</span><span class="params">()</span> </span>&#123;</span><br><span class="line">           <span class="keyword">return</span> desc;</span><br><span class="line">       &#125;</span><br><span class="line"> </span><br><span class="line">       <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setDesc</span><span class="params">(String desc)</span> </span>&#123;</span><br><span class="line">           <span class="keyword">this</span>.desc = desc;</span><br><span class="line">       &#125;</span><br><span class="line"> </span><br><span class="line">       <span class="meta">@Override</span></span><br><span class="line">       <span class="function"><span class="keyword">public</span> String <span class="title">toString</span><span class="params">()</span> </span>&#123;</span><br><span class="line">           <span class="keyword">return</span> <span class="string">"OrderStep&#123;"</span> +</span><br><span class="line">               <span class="string">"orderId="</span> + orderId +</span><br><span class="line">               <span class="string">", desc='"</span> + desc + <span class="string">'\''</span> +</span><br><span class="line">               <span class="string">'&#125;'</span>;</span><br><span class="line">       &#125;</span><br><span class="line">   &#125;</span><br><span class="line"> </span><br><span class="line">   <span class="comment">/**</span></span><br><span class="line"><span class="comment">    * 生成模拟订单数据</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">   <span class="function"><span class="keyword">private</span> List&lt;OrderStep&gt; <span class="title">buildOrders</span><span class="params">()</span> </span>&#123;</span><br><span class="line">       List&lt;OrderStep&gt; orderList = <span class="keyword">new</span> ArrayList&lt;OrderStep&gt;();</span><br><span class="line"> </span><br><span class="line">       OrderStep orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"创建"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111065L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"创建"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"付款"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103117235L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"创建"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111065L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"付款"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103117235L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"付款"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111065L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"完成"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"推送"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103117235L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"完成"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       orderDemo = <span class="keyword">new</span> OrderStep();</span><br><span class="line">       orderDemo.setOrderId(<span class="number">15103111039L</span>);</span><br><span class="line">       orderDemo.setDesc(<span class="string">"完成"</span>);</span><br><span class="line">       orderList.add(orderDemo);</span><br><span class="line"> </span><br><span class="line">       <span class="keyword">return</span> orderList;</span><br><span class="line">   &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="发送延迟消息"><a href="#发送延迟消息" class="headerlink" title="发送延迟消息"></a>发送延迟消息</h3><blockquote>
<p>现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18</p>
<p>private String messageDelayLevel = “1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”;</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">import</span> org.apache.rocketmq.client.producer.DefaultMQProducer;</span><br><span class="line"><span class="keyword">import</span> org.apache.rocketmq.common.message.Message;</span><br><span class="line"> </span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">ScheduledMessageProducer</span> </span>&#123;</span><br><span class="line">   <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">      <span class="comment">// 实例化一个生产者来产生延时消息</span></span><br><span class="line">      DefaultMQProducer producer = <span class="keyword">new</span> DefaultMQProducer(<span class="string">"ExampleProducerGroup"</span>);</span><br><span class="line">      <span class="comment">// 启动生产者</span></span><br><span class="line">      producer.start();</span><br><span class="line">      <span class="keyword">int</span> totalMessagesToSend = <span class="number">100</span>;</span><br><span class="line">      <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; totalMessagesToSend; i++) &#123;</span><br><span class="line">          Message message = <span class="keyword">new</span> Message(<span class="string">"TestTopic"</span>, (<span class="string">"Hello scheduled message "</span> + i).getBytes());</span><br><span class="line">          <span class="comment">// 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)</span></span><br><span class="line">          message.setDelayTimeLevel(<span class="number">3</span>);</span><br><span class="line">          <span class="comment">// 发送消息</span></span><br><span class="line">          producer.send(message);</span><br><span class="line">      &#125;</span><br><span class="line">       <span class="comment">// 关闭生产者</span></span><br><span class="line">      producer.shutdown();</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="发送批量消息"><a href="#发送批量消息" class="headerlink" title="发送批量消息"></a>发送批量消息</h3><blockquote>
<p>批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line">String topic = <span class="string">"BatchTest"</span>;</span><br><span class="line">List&lt;Message&gt; messages = <span class="keyword">new</span> ArrayList&lt;&gt;();</span><br><span class="line">messages.add(<span class="keyword">new</span> Message(topic, <span class="string">"TagA"</span>, <span class="string">"OrderID001"</span>, <span class="string">"Hello world 0"</span>.getBytes()));</span><br><span class="line">messages.add(<span class="keyword">new</span> Message(topic, <span class="string">"TagA"</span>, <span class="string">"OrderID002"</span>, <span class="string">"Hello world 1"</span>.getBytes()));</span><br><span class="line">messages.add(<span class="keyword">new</span> Message(topic, <span class="string">"TagA"</span>, <span class="string">"OrderID003"</span>, <span class="string">"Hello world 2"</span>.getBytes()));</span><br><span class="line"><span class="keyword">try</span> &#123;</span><br><span class="line">   producer.send(messages);</span><br><span class="line">&#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">   e.printStackTrace();</span><br><span class="line">   <span class="comment">//处理error</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="生产者常用设置参数"><a href="#生产者常用设置参数" class="headerlink" title="生产者常用设置参数"></a>生产者常用设置参数</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// 初始化 设置相关参数</span></span><br><span class="line">            defaultMQProducer = <span class="keyword">new</span> DefaultMQProducer(producerGroupName); </span><br><span class="line">             <span class="comment">// 设置nameSever地址</span></span><br><span class="line">            defaultMQProducer.setNamesrvAddr(namesrvAddr);</span><br><span class="line">             <span class="comment">// 实例名</span></span><br><span class="line">            defaultMQProducer.setInstanceName(instanceName);</span><br><span class="line">            <span class="comment">// 异步发送失败重发次数，默认2次</span></span><br><span class="line">            defaultMQProducer.setRetryTimesWhenSendAsyncFailed(<span class="number">10</span>);</span><br><span class="line">            <span class="comment">// 同步发送失败重发次数，默认2次</span></span><br><span class="line">            defaultMQProducer.setRetryTimesWhenSendFailed(<span class="number">10</span>);</span><br><span class="line">            <span class="comment">// 消息没有存储成功是否发送到另外一个broker</span></span><br><span class="line">            defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(<span class="keyword">true</span>);</span><br><span class="line">            <span class="comment">// 发送超时时间</span></span><br><span class="line">            defaultMQProducer.setSendMsgTimeout(<span class="number">5000</span>);</span><br><span class="line">            defaultMQProducer.start();</span><br></pre></td></tr></table></figure>

<h2 id="Consumer"><a href="#Consumer" class="headerlink" title="Consumer"></a>Consumer</h2><h3 id="Push模式"><a href="#Push模式" class="headerlink" title="Push模式"></a>Push模式</h3><blockquote>
<p>由于消息中间件主动地将消息推送给消费者，可以尽可能实时地将消息发送给消费者进行消费。但是消费者的处理消息的能力较弱时，比如(消费者端业务系统处理一条流程比较复杂，导致消费比较耗时)，而MQ不断地向消费者Push消息，消费者端缓冲区可能会溢出，导致异常。</p>
</blockquote>
<blockquote>
<p><strong>Push模式实际上在内部还是使用的Pull方式实现的，通过Pull不断地轮询Broker获取消息，当不存在新消息时，Broker端会挂起Pull请求，直到有新消息产生才取消挂起，返回新消息。</strong></p>
<p>需要返回消费状态，MQ内部维护Offset</p>
</blockquote>
<h3 id="Pull模式"><a href="#Pull模式" class="headerlink" title="Pull模式"></a>Pull模式</h3><blockquote>
<p>由消费者客户端主动向消息中间件（MQ消息服务器代理）拉取消息；采用Pull方式，如何设置Pull消息的频率需要重点去考虑，举个例子来说，可能1分钟内连续来了1000条消息，然后2小时内没有新消息产生（概括起来说就是<strong>“消息延迟与忙等待”</strong>）。如果每次Pull的时间间隔比较久，会增加消息的延迟，即消息到达消费者的时间加长，MQ中消息的堆积量变大；若每次Pull的时间间隔较短，但是在一段时间内MQ中并没有任何消息可以消费，那么会产生很多无效的Pull请求的RPC开销，影响MQ整体的网络性能；</p>
</blockquote>
<blockquote>
<p><strong>消息消费的偏移量需要Consumer端自己去维护</strong></p>
</blockquote>
<h3 id="集群模式-amp-广播模式"><a href="#集群模式-amp-广播模式" class="headerlink" title="集群模式&amp;广播模式"></a>集群模式&amp;广播模式</h3><blockquote>
<p>集群模式：一条消息，在同一个消费者组中，只有一个消费实例消费。</p>
<p>广播模式:   一条消息，在同一个消费者组中,全部消费实例都会消费。</p>
</blockquote>
<h3 id="消费者与队列关系"><a href="#消费者与队列关系" class="headerlink" title="消费者与队列关系"></a>消费者与队列关系</h3><p>消费者==队列数</p>
<blockquote>
<p>一个消费负责一个队列</p>
</blockquote>
<p>消费者&gt;队列数</p>
<blockquote>
<p>一个消费者负责一个队列，剩余消费者空闲</p>
</blockquote>
<p>消费者&lt;队列数</p>
<blockquote>
<p>队列数均摊所有消费者</p>
</blockquote>
<h3 id="消费重复问题"><a href="#消费重复问题" class="headerlink" title="消费重复问题"></a>消费重复问题</h3><p><strong>Redis</strong></p>
<blockquote>
<p>给消息分配一个全局id,只要消费过该消息，将&lt;id,message&gt;以K-V形式写入redis。消费者开始消费之前，先去redis中查询有没有消费记录即可</p>
</blockquote>
<h3 id="常用的命令"><a href="#常用的命令" class="headerlink" title="常用的命令"></a>常用的命令</h3><ol>
<li><p>查看所有topic</p>
<blockquote>
<p>sh mqadmin topicList -n localhost:9876</p>
</blockquote>
</li>
<li><p>查看clustername</p>
<blockquote>
<p>sh mqadmin clusterList -n localhost:9876</p>
</blockquote>
</li>
<li><p>查看所有消费组</p>
<blockquote>
<p>sh mqadmin consumerProgress -n localhost:9876</p>
</blockquote>
</li>
<li><p>查看消费组的发送、接收、阻塞情况</p>
<blockquote>
<p>sh mqadmin consumerProgress -g groupname -n localhost:9876</p>
</blockquote>
</li>
<li><p>删除topic</p>
<blockquote>
<p>sh mqadmin deleteTopic -n localhost:9876 -c cluster-bus -t topicname</p>
</blockquote>
</li>
<li><p>删除group</p>
<blockquote>
<p>sh mqadmin deleteSubGroup -c topicname -g groupname -n localhost:9876</p>
</blockquote>
</li>
<li><p>新增增或修改topic</p>
<blockquote>
<p>sh mqadmin updateTopic -c cluster-bus -n localhost:9876 -o true -p 6 -w 50 -r 50 -t topicName</p>
</blockquote>
</li>
<li><p>新增或修改消费组</p>
<blockquote>
<p>sh mqadmin updateSubGroup -n localhost:9876 -c cluster-bus -g groupname</p>
</blockquote>
</li>
<li><p>查看topic配置属性</p>
<blockquote>
<p>sh mqadmin topicRoute -n localhost:9876 -t topicname</p>
</blockquote>
</li>
<li><p>根据消息Key查询消息</p>
<blockquote>
<p>sh mqadmin queryMsgByKey -n localhost:9876 -t topicname -k key</p>
</blockquote>
</li>
<li><p>重置消费位点</p>
<blockquote>
<p>sh mqadmin resetOffsetByTime -n localhost:9876 -g groupname -t topicname -s now</p>
</blockquote>
</li>
</ol>

      
    </div>
    <footer class="article-footer">
      
        <div id="donation_div"></div>

<script src="/js/vdonate.js"></script>
<script>
var a = new Donate({
  title: '如果觉得我的文章对您有用，请随意打赏。您的支持将鼓励我继续创作!', // 可选参数，打赏标题
  btnText: '打赏支持', // 可选参数，打赏按钮文字
  el: document.getElementById('donation_div'),
  wechatImage: 'https://raw.githubusercontent.com/iTimeTraveler/iTimeTraveler.github.io/site/source/about/donate/images/WeChanQR.png',
  alipayImage: 'https://raw.githubusercontent.com/iTimeTraveler/iTimeTraveler.github.io/site/source/about/donate/images/AliPayQR.jpg'
});
</script>
      
      
      <div>
        <ul class="post-copyright">
          <li class="post-copyright-author">
          <strong>本文作者:  </strong>LeBron Tao
          </li>
          <li class="post-copyright-link">
          <strong>本文链接:  </strong>
          <a href="/2020/12/01/RocketMQ/" target="_blank" title="RocketMQ">https://jameslin23.gitee.io/2020/12/01/RocketMQ/</a>
          </li>
          <li class="post-copyright-license">
            <strong>版权声明:   </strong>
            本博客所有文章除特别声明外，均采用 <a rel="license" href="https://creativecommons.org/licenses/by-nc-nd/4.0/" target="_blank" title="Attribution-NonCommercial-NoDerivatives 4.0 International (CC BY-NC-ND 4.0)">CC BY-NC-ND 4.0</a>
            许可协议。转载请注明出处
          </li>
         
        </ul>
<div>
</div></div>
      
      
        
	<div id="comment">
		<!-- 来必力City版安装代码 -->
		<div id="lv-container" data-id="city" data-uid="MTAyMC8yOTQ4MS82MDQ5">
		<script type="text/javascript">
		   (function(d, s) {
		       var j, e = d.getElementsByTagName(s)[0];

		       if (typeof LivereTower === 'function') { return; }

		       j = d.createElement(s);
		       j.src = 'https://cdn-city.livere.com/js/embed.dist.js';
		       j.async = true;

		       e.parentNode.insertBefore(j, e);
		   })(document, 'script');
		</script>
		<noscript>为正常使用来必力评论功能请激活JavaScript</noscript>
		</div>
		<!-- City版安装代码已完成 -->
	</div>



      
      
        
  <ul class="article-tag-list"><li class="article-tag-list-item"><a class="article-tag-list-link" href="/tags/MQ/">MQ</a></li></ul>

      

    </footer>
  </div>
  
    
<nav id="article-nav">
  
    <a href="/2020/12/01/ngnix/" id="article-nav-newer" class="article-nav-link-wrap">
      <strong class="article-nav-caption">上一篇</strong>
      <div class="article-nav-title">
        
          Nginx
        
      </div>
    </a>
  
  
    <a href="/2020/11/28/java基础/" id="article-nav-older" class="article-nav-link-wrap">
      <strong class="article-nav-caption">下一篇</strong>
      <div class="article-nav-title">零散java知识补漏</div>
    </a>
  
</nav>

  
</article>

<!-- Table of Contents -->

  <aside id="toc-sidebar">
    <div id="toc" class="toc-article">
    <strong class="toc-title">文章目录</strong>
    
        <ol class="nav"><li class="nav-item nav-level-1"><a class="nav-link" href="#MQ有哪些作用"><span class="nav-number">1.</span> <span class="nav-text">MQ有哪些作用?</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#异步化提高系统性能"><span class="nav-number">1.1.</span> <span class="nav-text">异步化提高系统性能</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#系统解耦"><span class="nav-number">1.2.</span> <span class="nav-text">系统解耦</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#高并发削峰"><span class="nav-number">1.3.</span> <span class="nav-text">高并发削峰</span></a></li></ol></li><li class="nav-item nav-level-1"><a class="nav-link" href="#MQ技术选型"><span class="nav-number">2.</span> <span class="nav-text">MQ技术选型</span></a></li><li class="nav-item nav-level-1"><a class="nav-link" href="#RocketMQ核心原理"><span class="nav-number">3.</span> <span class="nav-text">RocketMQ核心原理</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#NameServer"><span class="nav-number">3.1.</span> <span class="nav-text">NameServer</span></a></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Broker"><span class="nav-number">3.2.</span> <span class="nav-text">Broker</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#CommitLog"><span class="nav-number">3.2.1.</span> <span class="nav-text">CommitLog</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#ConsumeQueue"><span class="nav-number">3.2.2.</span> <span class="nav-text">ConsumeQueue</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#异步刷盘和同步刷盘"><span class="nav-number">3.2.3.</span> <span class="nav-text">异步刷盘和同步刷盘</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#异步刷盘"><span class="nav-number">3.2.3.1.</span> <span class="nav-text">异步刷盘</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#同步刷盘"><span class="nav-number">3.2.3.2.</span> <span class="nav-text">同步刷盘</span></a></li></ol></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Producer"><span class="nav-number">3.3.</span> <span class="nav-text">Producer</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#同步发送（Sync）"><span class="nav-number">3.3.1.</span> <span class="nav-text">同步发送（Sync）</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#异步发送（async）"><span class="nav-number">3.3.2.</span> <span class="nav-text">异步发送（async）</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#直接发送-sendOneway"><span class="nav-number">3.3.3.</span> <span class="nav-text">直接发送(sendOneway)</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#顺序消息发送"><span class="nav-number">3.3.4.</span> <span class="nav-text">顺序消息发送</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#发送延迟消息"><span class="nav-number">3.3.5.</span> <span class="nav-text">发送延迟消息</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#发送批量消息"><span class="nav-number">3.3.6.</span> <span class="nav-text">发送批量消息</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#生产者常用设置参数"><span class="nav-number">3.3.7.</span> <span class="nav-text">生产者常用设置参数</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Consumer"><span class="nav-number">3.4.</span> <span class="nav-text">Consumer</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#Push模式"><span class="nav-number">3.4.1.</span> <span class="nav-text">Push模式</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Pull模式"><span class="nav-number">3.4.2.</span> <span class="nav-text">Pull模式</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#集群模式-amp-广播模式"><span class="nav-number">3.4.3.</span> <span class="nav-text">集群模式&amp;广播模式</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#消费者与队列关系"><span class="nav-number">3.4.4.</span> <span class="nav-text">消费者与队列关系</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#消费重复问题"><span class="nav-number">3.4.5.</span> <span class="nav-text">消费重复问题</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#常用的命令"><span class="nav-number">3.4.6.</span> <span class="nav-text">常用的命令</span></a></li></ol></li></ol></li></ol>
    
    </div>
  </aside>

</section>
        
      </div>
      
      <footer id="footer">
  

  <div class="container">
      	<div class="row">
	      <p> Powered by <a href="http://hexo.io/" target="_blank">Hexo</a> and <a href="https://github.com/iTimeTraveler/hexo-theme-hiker" target="_blank">Hexo-theme-hiker</a> </p>
	      <p id="copyRightEn">Copyright &copy; 2013 - 2021 JAVA All Rights Reserved.</p>
	      
	      
    		<p class="busuanzi_uv">
				访客数 : <span id="busuanzi_value_site_uv"></span> |  
				访问量 : <span id="busuanzi_value_site_pv"></span>
		    </p>
  		   
		</div>

		
  </div>
</footer>


<!-- min height -->

<script>
    var wrapdiv = document.getElementById("wrap");
    var contentdiv = document.getElementById("content");
    var allheader = document.getElementById("allheader");

    wrapdiv.style.minHeight = document.body.offsetHeight + "px";
    if (allheader != null) {
      contentdiv.style.minHeight = document.body.offsetHeight - allheader.offsetHeight - document.getElementById("footer").offsetHeight + "px";
    } else {
      contentdiv.style.minHeight = document.body.offsetHeight - document.getElementById("footer").offsetHeight + "px";
    }
</script>
    </div>
    <!-- <nav id="mobile-nav">
  
    <a href="/" class="mobile-nav-link">Home</a>
  
    <a href="/archives" class="mobile-nav-link">Archives</a>
  
    <a href="/categories" class="mobile-nav-link">Categories</a>
  
    <a href="/tags" class="mobile-nav-link">Tags</a>
  
    <a href="/about" class="mobile-nav-link">About</a>
  
</nav> -->
    

<!-- mathjax config similar to math.stackexchange -->

<script type="text/x-mathjax-config">
  MathJax.Hub.Config({
    tex2jax: {
      inlineMath: [ ['$','$'], ["\\(","\\)"] ],
      processEscapes: true
    }
  });
</script>

<script type="text/x-mathjax-config">
    MathJax.Hub.Config({
      tex2jax: {
        skipTags: ['script', 'noscript', 'style', 'textarea', 'pre', 'code']
      }
    });
</script>

<script type="text/x-mathjax-config">
    MathJax.Hub.Queue(function() {
        var all = MathJax.Hub.getAllJax(), i;
        for(i=0; i < all.length; i += 1) {
            all[i].SourceElement().parentNode.className += ' has-jax';
        }
    });
</script>

<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/mathjax/2.7.1/MathJax.js?config=TeX-AMS-MML_HTMLorMML">
</script>


  <link rel="stylesheet" href="/fancybox/jquery.fancybox.css">
  <script src="/fancybox/jquery.fancybox.pack.js"></script>


<script src="/js/scripts.js"></script>




  <script src="/js/dialog.js"></script>








	<div style="display: none;">
    <script src="https://s95.cnzz.com/z_stat.php?id=1260716016&web_id=1260716016" language="JavaScript"></script>
  </div>



	<script async src="//busuanzi.ibruce.info/busuanzi/2.3/busuanzi.pure.mini.js">
	</script>






  </div>

  <div class="modal fade" id="myModal" tabindex="-1" role="dialog" aria-labelledby="myModalLabel" aria-hidden="true" style="display: none;">
  <div class="modal-dialog">
    <div class="modal-content">
      <div class="modal-header">
        <h2 class="modal-title" id="myModalLabel">设置</h2>
      </div>
      <hr style="margin-top:0px; margin-bottom:0px; width:80%; border-top: 3px solid #000;">
      <hr style="margin-top:2px; margin-bottom:0px; width:80%; border-top: 1px solid #000;">


      <div class="modal-body">
          <div style="margin:6px;">
            <a data-toggle="collapse" data-parent="#accordion" href="#collapseOne" onclick="javascript:setFontSize();" aria-expanded="true" aria-controls="collapseOne">
              正文字号大小
            </a>
          </div>
          <div id="collapseOne" class="panel-collapse collapse" role="tabpanel" aria-labelledby="headingOne">
          <div class="panel-body">
            您已调整页面字体大小
          </div>
        </div>
      


          <div style="margin:6px;">
            <a data-toggle="collapse" data-parent="#accordion" href="#collapseTwo" onclick="javascript:setBackground();" aria-expanded="true" aria-controls="collapseTwo">
              夜间护眼模式
            </a>
        </div>
          <div id="collapseTwo" class="panel-collapse collapse" role="tabpanel" aria-labelledby="headingTwo">
          <div class="panel-body">
            夜间模式已经开启，再次单击按钮即可关闭 
          </div>
        </div>

        <div>
            <a data-toggle="collapse" data-parent="#accordion" href="#collapseThree" aria-expanded="true" aria-controls="collapseThree">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;关 于&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</a>
        </div>
         <div id="collapseThree" class="panel-collapse collapse" role="tabpanel" aria-labelledby="headingThree">
          <div class="panel-body">
            JAVA
          </div>
          <div class="panel-body">
            Copyright © 2021 LeBron Tao All Rights Reserved.
          </div>
        </div>
      </div>


      <hr style="margin-top:0px; margin-bottom:0px; width:80%; border-top: 1px solid #000;">
      <hr style="margin-top:2px; margin-bottom:0px; width:80%; border-top: 3px solid #000;">
      <div class="modal-footer">
        <button type="button" class="close" data-dismiss="modal" aria-label="Close"><span aria-hidden="true">×</span></button>
      </div>
    </div>
  </div>
</div>
  
  <a id="rocket" href="#top" class=""></a>
  <script type="text/javascript" src="/js/totop.js?v=1.0.0" async=""></script>
  
    <a id="menu-switch"><i class="fa fa-bars fa-lg"></i></a>
  
</body>
</html>